package com.learning.rocketmq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * ClassName: PushConsumer
 * Description: 由系统控制读取操作，收到消息后自动调用传入方法来处理
 * Date: 2018/12/26 14:45 【需求编号】
 *
 * @author Sam Sho
 * @version V1.0.0
 */
public class PushConsumer {
    public static void main(String[] args) throws MQClientException {

        // Consumer 的 GroupName 用于把多个 Consumer 组织到一起，提高鬓发处理能力。
        // GroupName 需要和消息模式（MessageModel）配合使用
        DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer("base_consumer_group");

        // NameServer 的地址与端口号。多个用分号隔开，达到消除单点故障的目的
        pushConsumer.setNamesrvAddr("10.0.64.106:9876;10.0.64.107:9876");

        // 设置消费位
        pushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        // CLUSTERING 模式，同一个 ConsumerGroup（GroupName相同）里的每个 Consumer 只消费所订阅消息的一部分。同一个 ConsumerGroup 里所有的 Consumer 消费的内容合起来才是所订阅Topic内容的整体，从而达到负载
        // BROADCASTING 模式，同一个 ConsumerGroup 里面的每个 Consumer 都能消费到所订阅 Topic 的全部消息。也就是一个消息会被多次分发，被多个 Consumer 消费
        pushConsumer.setMessageModel(MessageModel.CLUSTERING);

        // 设置 Topic 的名称用来标识消息类型。
        // 如果不需要消费某个topic 下的所有消息，可以通过指定消息的 Tag 进行消息过滤。null 或者 * 表示要消费全部消息
        pushConsumer.subscribe("base_topic", "*");


        pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                System.out.printf(Thread.currentThread().getName() + "收到新消息：" + msgs + "%n");
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        pushConsumer.start();
    }

    /**
     * DefaultMQPushConsumer 源码分析
     */
    public void test() {

        // 1- DefaultMQPushConsumer 主要功能是 DefaultMQPushConsumerImpl 类实现的，消息的处理在于方法 pullMessage 中的 PullCallback 中。在 PullCallback 的 onSuccess 函数中有个 switch 语句，根据从 Broker 返回的消息类型做相应的处理。

        // 2- DefaultMQPushConsumer 的源码中有很多 PullRequest 的语句，原因在于 "Push"是通过长轮询的方式达到的，长轮询方法既有pull的优点，又兼具Push方式的实时性

        // Push 方式是 Server 端接收到消息后，主动把消息推送给Client端，实时性高。但是有很多弊端：加大了 Server 的工作量影响 Server 的性能；其次Client的处理能力各不相同，Client 的状态不受Server控制，如果Client不能及时处理 Server 推送过来的消息，会造成各种潜在的问题。
        // Pull 方式是 Client 端循环地从 Server 端拉取消息，主动权在 Client 手中，自己拉取到一定数量后，处理妥当了在接着拉取。Pull 方式的问题时循环拉取的间隔不好设定，
        // 间隔太短就出在一个"盲等"的状态，浪费资源；每个pull的时间间隔太长，Server端有消息到来时，有可能没有被及时处理

        // 长轮询方式通过 Client 端和 Server 端的配合，达到既拥有 Pull 的优点，又能达到保证实时性的目的。

        // Pull 源码 设置 Broker 最长阻塞时间，默认是15秒，是Broker在没有新消息的时候才阻塞，有消息会立即返回
        // 长轮询服务端接收到消息请求后，如果队列里面没有新消息，并不急于返回，通过一个循环不断查看状态，每次 waitForRunning 一段时间（默认是5秒），然后再Check。
        // 默认情况下当 broker 一直没有新消息，第三次check的时候，等待时间超过 request 里面的 SuspendMaxTimeMills ，就返回空结果。在等待的过程中，Broker 收到新信息，会直接返回。
        // 长轮询的核心是 Broker 端 Hold 住客户单过来的请求一小段时间，在这个时间内有新消息到达，就利用现有的连接立刻返回消息给客户端。
        // 长轮询的主动权还是掌握在客户端手中，broker 即使有大量消息堆积，也不会主动推送给客户单

        // 长轮询方式的局限性，是在 hold 住 Consumer 请求时候需要占用资源，它适合在消息队列客户端连接数可控的场景

    }

    /**
     * DefaultMQPushConsumer 流量控制
     */
    public void test2() {
        // PushConsumer 使用多线程方式实现，有一个线程池
        // ProcessQueue
    }
}
